// readers/mysql_reader.go
package readers

import (
	"context"
	"crypto/tls"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"log/slog"
	"math/rand"
	"mig2kwdb/common"
	"os"
	"reflect"
	"strconv"
	"strings"
	"time"

	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/go-mysql-org/go-mysql/replication"
	MYSQL "github.com/zhanghaiyang9999/go-mysql-stdzlib"
)

type MySQLReader struct {
	db        *sql.DB
	connInfo  common.DBConnectionInfo
	dataBases []common.DatabaseConfig
	handler   common.ReplicateHandlerInterface
}

// NewMySQLReader 创建新的 MySQLReader 实例并初始化数据库连接
func NewMySQLReader(connInfo common.DBConnectionInfo) (*MySQLReader, error) {
	dsn := connInfo.User + ":" + connInfo.Password + "@tcp(" + connInfo.Ip + ":" + fmt.Sprint(connInfo.Port) + ")/?multiStatements=true&parseTime=true&maxAllowedPacket=1000000000&&readTimeout=3m&writeTimeout=2m&timeout=3m"
	if connInfo.Compress {
		dsn += "&compress=true"
	}
	if connInfo.SSL {
		dsn += "&tls=skip-verify"
	}
	db, err := sql.Open("mysql", dsn)

	return &MySQLReader{
		db:       db,
		connInfo: connInfo,
	}, err
}
func (r *MySQLReader) GetDB() *sql.DB {
	return r.db
}
func (r *MySQLReader) Close() {
	if r.db != nil {
		r.db.Close()
	}
}
func (r *MySQLReader) GetStartPos() (common.ReplicateStartPos, error) {
	var startPos common.ReplicateStartPos
	rows, err := r.db.Query("SHOW MASTER STATUS;")
	if err != nil {
		return startPos, err
	}

	defer rows.Close()
	cls, _ := rows.Columns()
	values := make([]interface{}, len(cls))
	for i := 0; i < len(cls); i++ {
		values[i] = &sql.NullString{}
	}

	rows.Next()
	err = rows.Scan(values...)
	if err != nil {
		log.Println(err)
		return startPos, err
	}

	if len(cls) > 2 {
		name := values[0].(*sql.NullString)
		if name.Valid {
			startPos.Name = name.String
		}
		pos := values[1].(*sql.NullString)
		if pos.Valid {

			var u32 uint64 = 0
			u32, err = strconv.ParseUint(pos.String, 10, 32)
			if err != nil {
				log.Println(err)
				return startPos, err
			}
			startPos.Position = (u32)
		}
	}
	str_sql := "show variables like 'server_id'"
	serverid_rows, serr := r.db.Query(str_sql)
	if serr == nil {
		defer serverid_rows.Close()
		serverid_rows.Next()
		var serverId string
		serverid_rows.Scan(&serverId, &startPos.ServerId)
	}
	return startPos, serr
}
func (r *MySQLReader) IsSyncedDB(dataBases []common.DatabaseConfig, dbName, tableName string) bool {

	for _, v := range dataBases {
		if !strings.EqualFold(dbName, v.Name) {
			continue
		}

		if len(tableName) < 1 {
			return true
		}
		for _, table := range v.Excludes {
			if table.Name == tableName {
				return false
			}
		}
		if len(v.Includes) == 0 {
			return true
		}
		for _, table := range v.Includes {
			if table.Name == tableName {
				return true
			}
		}
	}
	return false
}
func (r *MySQLReader) StartReplicate(ctx context.Context, startPos common.ReplicateStartPos) error {
	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
	serverId := uint32(rng.Intn(1000) + 1000)
	for {
		if serverId != (uint32)(startPos.ServerId) {
			break
		}
		serverId = uint32(rng.Intn(1000) + 1000)
	}
	tlsConfig := &tls.Config{InsecureSkipVerify: true}
	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
		Level: slog.LevelError, // 只显示 Error 及以上级别
	}))
	cfg := replication.BinlogSyncerConfig{
		ServerID:       serverId,
		Flavor:         "mysql",
		Host:           r.connInfo.Ip,
		Port:           uint16(r.connInfo.Port),
		User:           r.connInfo.User,
		Password:       r.connInfo.Password,
		RawModeEnabled: false,
		Logger:         logger,
	}
	if r.connInfo.SSL {
		cfg.TLSConfig = tlsConfig
	}
	syncer := replication.NewBinlogSyncer(cfg)
	if syncer == nil {

		return errors.New("NewBinlogSyncer failed")
	}
	var pos mysql.Position
	pos.Name = startPos.Name
	pos.Pos = uint32(startPos.Position)
	// Start sync with specified binlog file and position
	streamer, err := syncer.StartSync(pos)
	if err != nil {
		log.Println(err)
		return err
	}
	tablesMap := make(map[string][]common.KwdbColumn)
	defer syncer.Close()
	for {
		if common.IsStopped(ctx) {
			break
		}
		contx, cancel := context.WithTimeout(context.Background(), time.Second)
		ev, err := streamer.GetEvent(contx)
		cancel()

		if err == context.DeadlineExceeded {
			continue
		} else {
			if err != nil {
				if r.IsNetworkError(err) {
					continue
				} else {
					log.Println(err)
					return err
				}
			}
		}
		switch ev.Header.EventType {
		case replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2,
			replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2,
			replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2: //insert,update,delete
			event := ev.Event.(*replication.RowsEvent)
			if event.Rows == nil || len(event.Rows) == 0 {
				continue
			}
			dbName := string(event.Table.Schema)
			tableName := string(event.Table.Table)
			if !r.IsSyncedDB(r.dataBases, dbName, tableName) {
				continue
			}
			dbtb := dbName + "." + tableName
			cols, ok := tablesMap[dbtb]
			if !ok || len(cols) == 0 {
				columns, _ := r.handler.GetTableColumn(dbName, "", tableName)
				tablesMap[dbtb] = columns
				cols = columns
			}
			if ev.Header.EventType == replication.UPDATE_ROWS_EVENTv1 || ev.Header.EventType == replication.UPDATE_ROWS_EVENTv2 {
				//update
				_, err = r.handler.ReplicateRecords(dbName, "", tableName, cols, event.Rows, "update")

			} else if ev.Header.EventType == replication.DELETE_ROWS_EVENTv1 || ev.Header.EventType == replication.DELETE_ROWS_EVENTv2 {
				//delete
				_, err = r.handler.ReplicateRecords(dbName, "", tableName, cols, event.Rows, "delete")
			} else {
				//insert
				_, err = r.handler.ReplicateRecords(dbName, "", tableName, cols, event.Rows, "insert")

			}
			if err != nil {
				log.Println(err)
			}
		}
	}
	return nil
}
func (r *MySQLReader) GetDataBaseType() string {
	return "mysql"
}
func (r *MySQLReader) IsNetworkError(err error) bool {
	etype := reflect.TypeOf(err)
	name := etype.String()
	if strings.EqualFold(name, "*mysql.MySQLError") {
		mysqlerror := err.(*MYSQL.MySQLError)
		if mysqlerror.Number == 1053 { //shutdown
			return true
		} else {
			return false
		}
	}
	return true
}

// GetDatabaseTables retrieves a list of all tables in the specified database
func (r *MySQLReader) GetDatabaseTables(databaseName, schemaName string) ([]string, error) {
	query := `
		SELECT table_name 
		FROM information_schema.tables 
		WHERE table_schema = ? AND table_type = 'BASE TABLE'
		ORDER BY table_name
	`

	rows, err := r.db.Query(query, databaseName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var tables []string
	for rows.Next() {
		var tableName string
		if err := rows.Scan(&tableName); err != nil {
			return nil, err
		}
		tables = append(tables, tableName)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return tables, nil
}
func (r *MySQLReader) GetConditionRows(databaseName, schemaName, tableName string, columns []common.KwdbColumn, condition string) (common.RowsInterface, error) {
	orderby := ""
	keyNum := 0
	sql := "select "
	for idx, col := range columns {
		if idx > 0 {
			sql += ","
		}
		sql += " `" + col.Name + "`"
		if col.PrimaryKey.String == "YES" {
			if keyNum > 0 {
				orderby += ","
			}
			orderby += " `" + col.Name + "`"
			keyNum++
		}
	}
	// 如果没有主键，则使用唯一索引
	if orderby == "" {
		keyNum = 0
		for _, col := range columns {
			if col.Unique.String == "YES" {
				if keyNum > 0 {
					orderby += ","
				}
				orderby += " `" + col.Name + "`"
				keyNum++
			}
		}
	}
	sql += " from `" + databaseName + "`.`" + tableName + "`"
	if condition != "" {
		sql += " where " + condition
	}
	if orderby != "" {
		sql += " order by " + orderby
	}

	rows, err := r.db.Query(sql)
	return common.WrapSQLRows(rows), err
}

// GetDataBases retrieves a list of all databases in the MySQL instance
func (r *MySQLReader) GetDataBases() ([]string, error) {
	query := "SHOW DATABASES"
	rows, err := r.db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var databases []string
	for rows.Next() {
		var dbName string
		if err := rows.Scan(&dbName); err != nil {
			return nil, err
		}
		// 排除系统数据库
		if dbName != "information_schema" && dbName != "performance_schema" &&
			dbName != "mysql" && dbName != "sys" {
			databases = append(databases, dbName)
		}
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return databases, nil
}
func (r *MySQLReader) SetReplicateHandler(handler common.ReplicateHandlerInterface) {
	r.handler = handler
}
func (r *MySQLReader) SetDataBases(dataBases []common.DatabaseConfig) {
	r.dataBases = dataBases

}

func (r *MySQLReader) MapDataTypes(column []common.KwdbColumn) []common.KwdbColumn {
	for i := range column {
		switch strings.ToUpper(column[i].Type) {
		case "ENUM":
			column[i].Type = common.DataTypes.VARCHAR.Name
		case "YEAR", "TINYINT", "SMALLINT":
			column[i].Type = common.DataTypes.INT2.Name
		case "MEDIUMINT", "INT", "INTEGER":
			column[i].Type = common.DataTypes.INT4.Name
		case "TINYBLOB", "MEDIUMBLOB", "BLOB", "LONGBLOB", "BINARY", "VARBINARY":
			column[i].Type = common.DataTypes.BLOB.Name
		case "BIGINT":
			column[i].Type = common.DataTypes.INT8.Name
		case "FLOAT", "DOUBLE":
			column[i].Type = common.DataTypes.FLOAT8.Name
		case "SMALLMONEY", "MONEY", "DECIMAL", "NUMERIC":
			column[i].Type = common.DataTypes.DECIMAL.Name
		case "DATETIME":
			column[i].Type = common.DataTypes.TIMESTAMP.Name
		case "TEXT", "TINYTEXT", "MEDIUMTEXT", "LONGTEXT":
			column[i].Type = common.DataTypes.TEXT.Name

		default:
		}
	}
	return column

}

func (r *MySQLReader) GetTableDefinition(databaseName, schemaName, tableName string) (common.TableDefinition, error) {
	var err error
	var def common.TableDefinition
	def.DatabaseName = databaseName
	def.SchemaName = schemaName
	def.TableName = tableName
	def.Columns, err = r.GetTableColumn(databaseName, schemaName, tableName)
	if err != nil {
		log.Println(err)
		return def, err
	}

	return def, nil
}

// GetTableIndexes retrieves a list of all indexes for a given table
func (r *MySQLReader) GetTableIndexes(databaseName, schemaName, tableName string) ([]common.IndexInfo, error) {
	query := `
		SELECT 
			INDEX_NAME,
			NON_UNIQUE,
			COLUMN_NAME,
			SEQ_IN_INDEX,
			COLLATION
		FROM information_schema.statistics
		WHERE table_schema = ? AND table_name = ?
		ORDER BY INDEX_NAME, SEQ_IN_INDEX
	`

	rows, err := r.db.Query(query, databaseName, tableName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	// 使用 map 来组织索引信息
	indexMap := make(map[string]*common.IndexInfo)

	for rows.Next() {
		var indexName string
		var nonUnique int
		var columnName string
		var seqInIndex int
		var collation sql.NullString

		err := rows.Scan(&indexName, &nonUnique, &columnName, &seqInIndex, &collation)
		if err != nil {
			return nil, err
		}

		// 如果索引不存在，创建新的索引信息
		if _, exists := indexMap[indexName]; !exists {
			indexInfo := &common.IndexInfo{
				Name:      indexName,
				IsUnique:  nonUnique == 0,
				IsPrimary: indexName == "PRIMARY",
				Type:      "INDEX",
			}
			if indexName == "PRIMARY" {
				indexInfo.Type = "PRIMARY"
			} else if nonUnique == 0 {
				indexInfo.Type = "UNIQUE"
			}
			indexMap[indexName] = indexInfo
		}

		// 添加列信息到索引中
		indexColumn := common.IndexColumn{
			Name:      columnName,
			Position:  seqInIndex,
			SortOrder: "ASC",
		}

		// 根据排序规则确定排序顺序
		if collation.Valid && collation.String == "D" {
			indexColumn.SortOrder = "DESC"
		}

		indexMap[indexName].Columns = append(indexMap[indexName].Columns, indexColumn)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	// 将 map 转换为 slice
	var indexes []common.IndexInfo
	for _, indexInfo := range indexMap {
		indexes = append(indexes, *indexInfo)
	}

	return indexes, nil
}

// GetFields retrieves column information for a specific table in a MySQL database
func (r *MySQLReader) GetTableColumn(databaseName, schemaName, tableName string) ([]common.KwdbColumn, error) {
	query := `
		SELECT 
			c.column_name,
			c.ordinal_position,
			UPPER(c.data_type),
			COALESCE(c.character_maximum_length, 0),
			COALESCE(c.numeric_precision, 0),
			COALESCE(c.numeric_scale, 0),
			c.is_nullable,
			COALESCE(c.column_default, ''),
			CASE WHEN c.column_key = 'PRI' THEN 'YES' ELSE 'NO' END,
			CASE WHEN c.column_key = 'UNI' THEN 'YES' ELSE 'NO' END,
			COALESCE(c.column_comment, ''),
			CASE WHEN c.extra = 'auto_increment' THEN 'YES' ELSE 'NO' END,
			COALESCE(c.character_set_name, ''),
			COALESCE(c.collation_name, '')
		FROM information_schema.columns c
		WHERE c.table_schema = ? AND c.table_name = ?
		ORDER BY c.ordinal_position
	`

	rows, err := r.db.Query(query, databaseName, tableName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var columns []common.KwdbColumn
	for rows.Next() {
		var col common.KwdbColumn

		// 将查询结果直接扫描到 KwdbColumn 结构体中
		err := rows.Scan(
			&col.Name,
			&col.Position,
			&col.Type,
			&col.Length,
			&col.Precision,
			&col.Scale,
			&col.Nullable,
			&col.DefaultValue,
			&col.PrimaryKey,
			&col.Unique,
			&col.Comment,
			&col.AutoIncrement,
			&col.Charset,
			&col.Collation,
		)
		if err != nil {
			return nil, err
		}

		columns = append(columns, col)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return r.MapDataTypes(columns), nil
}
